﻿using Org.Apache.Rocketmq;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;

namespace DSQTransferService.Public
{
    public class RocketMQ
    {
        private readonly string endpoints = "172.17.212.221:8081";
        private readonly string topic;
        private readonly string tag;

        public delegate void RecivceMHandlel(string Message);//传递数据
        public event RecivceMHandlel RecivceM;

        public RocketMQ(string endpoints,string topic, string tag)
        {
            this.endpoints = endpoints;
            this.topic = topic;
            this.tag = tag;
        }
        public  async Task QuickStart()
        {
            // Enable the switch if you use .NET Core 3.1 and want to disable TLS/SSL.
            // AppContext.SetSwitch("System.Net.Http.SocketsHttpHandler.Http2UnencryptedSupport", true);
            //const string accessKey = "yourAccessKey";
            //const string secretKey = "yourSecretKey";

            //// Credential provider is optional for client configuration.
            //var credentialsProvider = new StaticSessionCredentialsProvider(accessKey, secretKey);//验证暂时不用
            var clientConfig = new ClientConfig.Builder()
                .SetEndpoints(endpoints)
                .Build();

            // Add your subscriptions.
            const string consumerGroup = "yourConsumerGroup";
             
            var subscription = new Dictionary<string, FilterExpression>
                { { topic, new FilterExpression(tag) } };
            // In most case, you don't need to create too many consumers, single pattern is recommended.
            var simpleConsumer = await new SimpleConsumer.Builder()
                .SetClientConfig(clientConfig)
                .SetConsumerGroup(consumerGroup)
                .SetAwaitDuration(TimeSpan.FromSeconds(15))
                .SetSubscriptionExpression(subscription)
                .Build();

            while (true)
            {
                var messageViews = await simpleConsumer.Receive(16, TimeSpan.FromSeconds(15));
                foreach (var message in messageViews)
                {
                    LogHelper.Instance.Info($"Received a message, topic={message.Topic},具体内容{Encoding.UTF8.GetString((message.Body))}");
                    Console.WriteLine($"Received a message, topic={message.Topic},具体内容{Encoding.UTF8.GetString((message.Body))}");
                    RecivceM.Invoke(Encoding.UTF8.GetString((message.Body)));
                    await simpleConsumer.Ack(message);
                    LogHelper.Instance.Info($"Message is acknowledged successfully, message-id={message.Topic}");
                    Console.WriteLine($"Message is acknowledged successfully, message-id={message.Topic}");                 
                }
            }
            // Close the simple consumer if you don't need it anymore.
            // await simpleConsumer.DisposeAsync();
        }

    }
}
